package com.niodata.dt.fs.us3;

import cn.ucloud.ufile.api.object.DownloadFileApi;
import cn.ucloud.ufile.bean.DownloadFileBean;
import cn.ucloud.ufile.bean.ObjectProfile;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.UUID;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.log4j.Logger;

public class Us3InputStream extends FSInputStream {

  private static final Logger logger = Logger.getLogger(Us3InputStream.class);
  private final ObjectProfile profile;
  private final DownloadFileApi api;
  //16mb local buffer
  private final int bufferSize = 1024 * 1024 * 16;
  private final int halfSize = bufferSize / 2;
  private final byte[] buffer;
  private final String localTmpDir;
  private int bufferOffset = 0;
  private int bufferLimit = 0;
  private long remoteReadSize = 0;
  private long readSize;


  /**
   * us3 inputstream.
   *
   * @param profile profile
   * @param api api
   */
  public Us3InputStream(ObjectProfile profile, DownloadFileApi api) {
    this.profile = profile;
    this.api = api;
    localTmpDir = System.getProperty("user.dir") + "/.us3tmp/";
    File dir = new File(localTmpDir);
    if (!dir.exists()) {
      dir.mkdirs();
    }
    if (profile.getContentLength() <= bufferSize) {
      buffer = new byte[(int) profile.getContentLength()];
    } else {
      buffer = new byte[bufferSize];
    }
    initBuffer();
    logger.info(String.format("open us3 %s,%s,size=%d",
          profile.getBucket(),
          profile.getKeyName(),
          profile.getContentLength()));

  }

  @Override
  public int read() throws IOException {
    fillBufferIfNeeded();
    if (bufferOffset == bufferLimit) {
      return -1;
    }
    int b = -1;
    if (bufferOffset < buffer.length) {
      b = buffer[bufferOffset++];
    }
    readSize++;
    return b;
  }

  @Override
  public int read(byte[] bytes) throws IOException {
    fillBufferIfNeeded();
    if (bufferOffset == bufferLimit) {
      return -1;
    }
    int len = bufferLimit - bufferOffset < bytes.length ? bufferLimit - bufferOffset : bytes.length;
    System.arraycopy(buffer, bufferOffset, bytes, 0, len);
    bufferOffset = bufferOffset + len;
    readSize += len;
    return len;
  }

  @Override
  public int read(byte[] bytes, int off, int len) throws IOException {
    fillBufferIfNeeded();
    if (bufferOffset == bufferLimit) {
      return -1;
    }
    len = bufferLimit - bufferOffset > len ? len : bufferLimit - bufferOffset;
    System.arraycopy(buffer, bufferOffset, bytes, off, len);
    bufferOffset = bufferOffset + len;
    readSize += len;
    return len;
  }

  private void fillBufferIfNeeded() throws IOException {
    if (remoteReadSize < profile.getContentLength() && bufferLimit - bufferOffset < halfSize) {
      File tmpFile = new File(localTmpDir + '/' + UUID.randomUUID());
      try {
        int readSize = bufferSize - bufferOffset;
        long max = remoteReadSize + readSize;
        if (max > profile.getContentLength()) {
          max = profile.getContentLength();
        }
        max = max - 1;
        logger.info(String.format("fetch us3 %s,%s, from %d to %d",
              profile.getBucket(),
              profile.getKeyName(),
              remoteReadSize - 1, max));
        DownloadFileBean fileBean = this.api.which(profile).withinRange(remoteReadSize, max)
              .saveAt(localTmpDir, tmpFile.getName())
              .execute();
        byte[] bytes = Files.readAllBytes(fileBean.getFile().toPath());
        System.arraycopy(buffer, bufferOffset, buffer, 0, bufferLimit - bufferOffset);
        System.arraycopy(bytes, 0, buffer, bufferLimit - bufferOffset, bytes.length);
        bufferLimit = bytes.length + bufferLimit - bufferOffset;
        bufferOffset = 0;
        remoteReadSize = remoteReadSize + bytes.length;
      } catch (Exception e) {
        throw new RuntimeException(e);
      } finally {
        tmpFile.delete();
      }
    }
  }

  private void initBuffer() {
    long off = 0;
    long max =
          profile.getContentLength() > bufferSize ? bufferSize - 1 : profile.getContentLength() - 1;
    File tmpFile = new File(localTmpDir + '/' + UUID.randomUUID());
    try {
      DownloadFileBean fileBean = this.api.which(profile).saveAt(localTmpDir,
            tmpFile.getName()).withinRange(off, max).execute();
      byte[] bytes = Files.readAllBytes(fileBean.getFile().toPath());
      System.arraycopy(bytes, 0, buffer, 0, bytes.length);
      bufferLimit = bytes.length;
      remoteReadSize = bytes.length;
      bufferOffset = 0;
    } catch (Exception e) {
      throw new RuntimeException(e);
    } finally {
      tmpFile.delete();
    }
  }

  @Override
  public void seek(long pos) throws IOException {
    if (pos >= profile.getContentLength()) {
      throw new IOException("End of file");
    }
    bufferLimit = 0;
    bufferOffset = 0;
    fillBufferIfNeeded();
    readSize = pos;
  }

  @Override
  public long getPos() throws IOException {
    return readSize;
  }

  @Override
  public boolean seekToNewSource(long targetPos) throws IOException {
    return false;
  }

  @Override
  public int available() throws IOException {
    return (int) (profile.getContentLength() - readSize);
  }
}
